package rpcclient

import (
	"cloud_worm/util/unet/rpc"
	"crypto/tls"
	"net"
	"time"
	"github.com/golang/glog"
	"fmt"
	"sync"
	"io"
)

type Pusher interface {
	GetMethodName() string
	GetArgs() interface{}
	GetReply() interface{}

	IsNeedAgain() bool

	GetTryCount() int
	AddTryCount()
}

type WebClient struct {
	PushOneC		chan Pusher

	pushTwoC		chan Pusher			// 一级、延时重试的缓存区
	pushThreeC		chan Pusher			// 二级、延时重试的缓存区

	ServerAddr 		string
	client 			*rpc.Client
}

var PushClient *WebClient

func init() {
	PushClient = &WebClient{
		// 缓存队列长度为：100W
		PushOneC: make(chan Pusher, 1024 * 100 * 10),
		pushTwoC: make(chan Pusher, 1024 * 100),
		pushThreeC: make(chan Pusher, 1024 * 100),
	}
	return
}

func (w *WebClient) connect() error {
	if w.client != nil {
		w.client.Close()
		w.client = nil
	}

	config := &tls.Config{InsecureSkipVerify: true}
	d := &net.Dialer{Timeout: 10 * time.Second}
	conn, err := tls.DialWithDialer(d, "tcp", w.ServerAddr, config)
	if err != nil {
		return err
	}
	glog.Warningf("rpc client reconnect: %s", w.ServerAddr)

	option := rpc.ClientOption{
		Debug: false,
		ReadTimeout: 30 * time.Second,
		WriteTimeout: 5 * time.Second,
		HeartbeatInterval: 20 * time.Second,
	}
	w.client = rpc.NewClient(conn, option)
	return nil
}

func (w *WebClient) call(item Pusher) (again bool, err error) {
	if w.client == nil {
		err = w.connect()
		if err != nil {
			again = true
			return
		}
	}

	err = w.client.Call(item.GetMethodName(), item.GetArgs(), item.GetReply())
	if err != nil {
		glog.Errorf("Method: %s, Args: %v, Error: %s", item.GetMethodName(), item.GetArgs(), err.Error())
		if err == rpc.ErrShutdown || err == io.ErrUnexpectedEOF {
			err = w.connect()
			if err != nil {
				again = true
				return
			}

			again = true
			err = nil
			return
		} else {
			again = false
			return
		}
	}

	again = false
	err = nil
	return
}

func (w *WebClient) Loop(wg *sync.WaitGroup) error {
	if w.ServerAddr == "" {
		return fmt.Errorf("input server addr error")
	}

	// PushOneC 的消费协程
	go func(w *WebClient, wg *sync.WaitGroup) {
		wg.Add(1)
		for item := range w.PushOneC {
			again, err := w.call(item)
			if err != nil {
				glog.Errorf("call error: %s", err.Error())
			}

			if again && item.IsNeedAgain() {
				// 重试一次，放到 pushTwoC
				glog.Errorf("Method: %s, Args: %v, Try Count: %d", item.GetMethodName(), item.GetArgs(), item.GetTryCount())
				item.AddTryCount()
				w.pushTwoC <- item
				continue
			}
		}
		wg.Done()
		return
	}(w, wg)

	// pushTwoC 的消费协程
	go func(w *WebClient, wg *sync.WaitGroup) {
		wg.Add(1)
		for item := range w.pushTwoC {
			again, err := w.call(item)
			if err != nil {
				glog.Errorf("call error: %s", err.Error())
			}

			if again && item.IsNeedAgain() {
				// 重试第二次，放到 pushThreeC
				glog.Errorf("Method: %s, Args: %v, Try Count: %d", item.GetMethodName(), item.GetArgs(), item.GetTryCount())
				item.AddTryCount()
				w.pushThreeC <- item
				continue
			}
			time.Sleep(2 * time.Second)
		}
		wg.Done()
		return
	}(w, wg)

	// pushThreeC 的消费协程
	go func(w *WebClient, wg *sync.WaitGroup) {
		wg.Add(1)
		for item := range w.pushThreeC {
			_, err := w.call(item)
			if err != nil {
				glog.Errorf("Method: %s, Args: %v, error: %s", item.GetMethodName(), item.GetArgs(), err.Error())
			}

			// 不管结果怎样，不继续处理
			time.Sleep(4 * time.Second)
		}
		wg.Done()
		return
	}(w, wg)

	return nil
}